package example;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.Function;

import java.io.IOException;
import java.sql.Timestamp;

/**
 * 行为数据清洗类
 * <p>
 * 该类用于对用户行为数据进行清洗处理，包括字段格式化、时间解析和数据过滤等操作。
 * 输入数据为CSV格式的行为日志，输出为清洗后的结构化数据。
 * </p>
 */
public class BehaviorDataCleaning {

    private static String in = "/user/hive/warehouse/ods.db/behavior.csv";
    private static String out = "/user/hive/warehouse/dwd.db/behavior";

    /**
     * 主函数，执行数据清洗流程
     *
     * @param args 命令行参数
     * @throws IOException 文件系统操作异常时抛出
     */
    public static void main(String[] args) throws IOException {

        System.setProperty("HADOOP_USER_NAME","root");

        SparkConf conf = new SparkConf().setAppName("BehaviorDataCleaning").setMaster("local[*]");

        JavaSparkContext context = new JavaSparkContext(conf);

        /**
         * 检查输出路径是否存在，如果存在则删除
         */
        checkExistsFile(context.hadoopConfiguration(), new Path(out));

        JavaRDD<String> inRDD = context.textFile(in);

        /**
         * 对原始数据进行清洗转换：
         * - 分割字段并验证字段数量
         * - 解析时间戳为标准时间格式
         * - 提取日期和小时信息
         * - 过滤指定日期范围外的数据
         * - 过滤无效或格式错误的数据行
         */
        JavaRDD<String> mapRDD = inRDD.map(new Function<String, String>() {
            @Override
            public String call(String s) throws Exception {

                String[] fields = s.split(",");

                /**
                 * 验证字段长度是否符合预期（5个字段）
                 */
                if (fields.length == 5) {
                    String userId = fields[0];
                    String itemId = fields[1];
                    String categoryId = fields[2];
                    String type = fields[3];
                    String timestamp = fields[4];

                    /**
                     * 将时间戳转换为标准时间格式
                     */
                    Timestamp tmpTime = new Timestamp(Long.valueOf(timestamp) * 1000L);

                    String times = tmpTime.toString().substring(0, 19);

                    /**
                     * 提取日期部分（年-月-日）
                     */
                    String day = tmpTime.toString().substring(0, 10);

                    /**
                     * 提取小时部分
                     */
                    String hour = tmpTime.toString().substring(11, 13);
                    /**
                     * 过滤日期范围，仅保留2017-11-25至2017-12-03之间的数据
                     */
                    if(day.compareTo("2017-11-25")>=0 && day.compareTo("2017-12-03") <=0){
                        return userId + "," + itemId + "," + categoryId + "," + type + "," + times + "," + day + "," + hour;
                    }else{
                        return null;
                    }
                } else {
                    return null;
                }
            }
        }).filter(row -> null != row);  // 清洗掉null值

        mapRDD.saveAsTextFile(out);
        context.close();
    }

    /**
     * 检查文件是否存在，存在则删除
     *
     * @param conf Hadoop配置对象
     * @param path 要检查的文件路径
     * @throws IOException 文件系统操作异常时抛出
     */
    private static void checkExistsFile(Configuration conf, Path path) throws IOException {

        FileSystem fileSystem = FileSystem.get(conf);
        if(fileSystem.exists(path)){
            fileSystem.delete(path, true);
            fileSystem.close();
        }
    }
}
